<?php namespace xlx; 
use Workerman\Connection\TcpConnection;
use \Workerman\Events\EventInterface;

class Danboo{
    static $addr = 'tcp://127.0.0.1:55851';
    static $protocol =  '\xlx\Protocols\Frame';

    ////////////// service ///////////////// 


    /** 
     * .server
     * ```
     * [ 
     *   'coids'=>[ 'coid'=>['conn'=>TcpConnection ... ] ] //coid 所声明进程/关联重启  
     *   'alias'=>[ 'atom'=>'coid'  ]      //别名 查找唯一 coid 
     *   'link'=>[ 'coid'=>[ 'coid'=>OPTION ] ]      //coid 析构时群发 EXIT .
     * ]
     * ```
     * .client
     * ```
     * [
     *    'TRAPEXIT'=>[ 'coid'=>bool ]       //接收EXIT后的行为
     * ]
     * ```
     */
    static $data_server=['workers'=>[],'coids'=>[],'alias'=>[],'link'=>[],'cmd'=>[],]; 
    static $gen_coid = 0;


    public static function start()
    {
        //channel process
        \xlx\Process::fork(function () {
            yield Danboo::startServer();  //消息监听 tcp
            yield Danboo::startSupPool(); //进程池 actor
        }); 
    }
    public static function startServer()
    {
        if (\xlx\Process::checkPort(self::$addr)) {
            exit("server started\n");
        }
        $event = \Workerman\Worker::$globalEvent;
        $socket = \stream_socket_server(self::$addr, $errno, $errstr);
        $event->add($socket, EventInterface::EV_READ, function ($socket) {
            $socket = \stream_socket_accept($socket, 0, $addr);
            $client = new class($socket, $addr) extends TcpConnection{
                public function __debugInfo()
                {
                    return ['debuginfo'=>$this->_remoteAddress];
                }
                public function __toString()
                {
                    return $this->_remoteAddress;
                }
            };
            $client->protocol = self::$protocol;
            $client->onMessage = [self::class,'onServerMessage'];
            $client->onClose = [self::class,'onServerClose'];
 
        });

        $event->add(0.2, EventInterface::EV_TIMER, function () {
            self::$data_server['time'] = \microtime(true);
            //file_put_contents('nil', print_r(self::$data_server['coids'],true));
        });
        //self::$selfname =  '0000000000.0000';//\stream_socket_get_name( self::$server->getSocket(), false);
        // $event->add(1, EventInterface::EV_TIMER, function () { 
        //     foreach (self::$data_server['coids'] as $coid => $value) { 
        //         if(isset($value)){
        //             $value->send(['PING']);
        //         }
        //     }
        //     self::$data_server['ping']=[];
        // });
    }
    public static function startSupPool()
    { 
        $pid = yield \xlx\AChannel::make($actor,$stack);  
        $actor->trap_exit(true);  
        $actor->register('danboo.server'); 
        $actor->send($pid,['POOLS_START']);
        while(1){
            $item = yield $actor->receive(function($item){return $item;}); 
            switch ($item[0]) {
                case 'EXIT':  
                    unset(self::$data_server['workers'][$item[1]]); 
                case 'POOLS_START':
                    while(count(self::$data_server['workers'])<3){ 
                        \xlx\Process::fork([self::class,'startPoolChild'],$pid); 
                        $worker_coid = yield $actor->receive();
                        $actor->monitor($worker_coid);
                        $conn = \xlx\Danboo::$data_server['coids'][$worker_coid]; 
                        self::$data_server['workers'][$worker_coid] = $conn; 
                    } 
                    break; 
                case 'SUP_START'://0:TOKEN 1:INIT_FN 2:RETURN_COID 
                    $coid = self::$gen_coid++;
                    $coid = 'danboo.server:'.$coid; 
                    if(is_string($item[1]) && substr($item[1],0,38)
                        =='C:32:"Opis\Closure\SerializableClosure'){
                        $item[1] = unserialize($item[1]);
                    }  
                    $sup = self::$data_server['coids'][$coid] = new \xlx\DbSupervisor( $coid, $item); 
                    $sup->restart()->then(function()use($actor,$item,$coid){
                        $actor->send($item[2],$coid);  
                    });
                    break;
            }
        } 
    }
    public static function startPoolChild($event,$pid){
        yield \xlx\AChannel::make($actor,$stack);  
        $actor->send($pid,$actor->self()); 
        while(1){
            yield $actor->receive(function($item){
                switch ($item[0]) {
                    case 'SPAWN': //0:TOKEN 1:FN 2:ARGS , 3:FROM ,4:TICK
                        \xlx\co(function()use($item){ 
                            $self = yield \xlx\AChannel::make($actor,$stack);  
                            self::$conn_server->send([$item[4] ,$item[3], $self ]);  
                            yield \xlx\Process::delay(1); 
                            if(is_string($item[1]) && substr($item[1],0,38)=='C:32:"Opis\Closure\SerializableClosure'){
                                $item[1] = unserialize($item[1]);
                            }  
                            yield \call_user_func($item[1], $actor, $item[2] );   
                        });
                        return true; 
                    case 'EXIT':
                        $a=1;
                        break;
                    default: 
                        break;
                }
            });
            
        } 

    }
 
    static $conn_server;// TcpConnection 
    public static function connectServer()
    {
        $socket = \stream_socket_client(self::$addr, $errno, $errstr) or exit('??');
        self::$conn_server = new TcpConnection($socket);

        self::$conn_server->protocol = self::$protocol;
        self::$conn_server->onMessage = [self::class,'onClientMessage'];
        self::$conn_server->onClose = [self::class,'onClientClose'];
        //self::$selfname = \stream_socket_get_name( self::$conn_server->getSocket(), true);
        
        //echo "\n\n\n[".self::$selfname."]\n\n\n";
        //$event = \Workerman\Worker::$globalEvent;
        // $event->add(1, EventInterface::EV_TIMER, function () {
        //     self::$conn_server->send(['PING',[
        //         'time'=>time(),  
        //     ]]);
        // });
    }


    public static function onServerMessage($conn, $data)
    {
        // array_unshift(self::$data_server['cmd'], json_encode($data) );
        // if(count(self::$data_server['cmd'])>15){
        //     array_pop(self::$data_server['cmd']);
        // }
 
        $test = &self::$data_server;
        switch ($data[0]) {
            case 'PING':
                //self::$data_server['ping'][$conn->id] = $data[1];
                break;
            case 'NEW'://0:TOKEN 1:STACK_HASH 2:TICK
                $gen = self::$gen_coid++;
                $gen =  $conn->getRemotePort().':'.$gen;
                self::$data_server['coids'][$gen] = $conn;
                self::$data_server['coids'][$gen]->coid = $gen;   
                self::$data_server['coids'][$gen]->stack = $data[1];   
                $conn->send([$data[0] ,$data[2], $gen ]);
                break;
            // case 'CONN'://0:TOKEN,1:STACK_HASH, 2:COID
            //     if(isset(self::$data_server['coids'][$data[2]])){
            //         self::$data_server['coids'][$data[2]]->coid = $data[2];   
            //     }else{
            //         // print_r(self::$data_server['coids']);
            //         // print_r($data);
            //         throw new Exception("Error Processing Request", 1); 
            //     }
            //     break;
            case 'CLOSE'://0:TOKEN, 1:COID  
                if (isset(self::$data_server['link'][$data[1]])) { 

                    //删除确认
                    //$conn->send(['EXIT',$data[1],$proc->stack??null]);
                    foreach (self::$data_server['link'][$data[1]] as $to=>$class) {
                        //print_r([self::$data_server['link'],self::$data_server['coids'],$data]);

                        if(isset(self::$data_server['coids'][$to])){
                            $proc = self::$data_server['coids'][$to]; 
                            $proc->send([  'EXIT',$to ,$proc->stack??null,$data[1] ]); 
                        }else{ 

                            echo "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
                        }
                        // if(isset($proc->stack) && isset($proc)){
                        // }
                        // if(isset($proc['childs']) && isset($proc['onexit'])){ 
                        //     unset(self::$data_server['link'][$to][$data[1]]);
                        //     $proc['onexit']($to,$class,$proc['option']);
                        // }
                    }
                    //unset(self::$data_server['link'][$data[1]]);
                } 
                if (isset(self::$data_server['alias'])) { 
                    foreach (self::$data_server['alias'] as $atom => $coid) {
                        if ($coid == $data[1]) {
                            unset(self::$data_server['alias'][$atom]);
                        }
                    }   
                }
                //if(isset(self::$data_server['coids'][$data[1]])){
                    unset(self::$data_server['coids'][$data[1]]);
                //}
                break; 
            case 'SPAWN'://0:TOKEN, 1:FUNC 2:ARGS , 3:FROM ,4:TICK
                \xlx\co(function()use($conn,$data){
                    while(1){
                        if($pid = array_rand(self::$data_server['workers'])){
                            if(isset(self::$data_server['coids'][$pid])){
                                self::$data_server['coids'][$pid]->send(['SEND',$pid,$data]);  
                            }
                            return;
                        }
                        yield \xlx\Process::delay(0.3);
                    } 
                });
                break; 
            case 'SEND': //0:TOKEN, 1:TO, 2:DATA, 3:FROM ,4:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $data[1] = (self::$data_server['alias'][$data[1]]);
                }
                if (isset(self::$data_server['coids'][$data[1]])) {
                    //$conn->send([$data[4] ,$data[3], 'ok' ]);
                    self::$data_server['coids'][$data[1]]->send($data);
                } else {
                    //$conn->send([$data[4] ,$data[3], 'err','econnect send' ]);
                    //self::$data_server['err'][$conn->id]++;
                }
                break;
            case 'WHEREIS'://0:TOKEN, 1:ATOM, 2:FROM ,3:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $coid = (self::$data_server['alias'][$data[1]]);
                }
                $conn->send([$data[3] ,$data[2],$coid??'']);
                break;
            case 'REGISTER'://0:TOKEN, 1:ATOM, 2:FROM ,3:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $conn->send([$data[3] ,$data[2],'err','isset']);
                    //self::$data_server['err'][$conn->id]++;
                } else {
                    self::$data_server['alias'][$data[1]] = $data[2];
                    $conn->send([$data[3] ,$data[2],'ok' ]);
                }
                break;
            case 'UNREGISTER'://0:TOKEN, 1:ATOM, 2:FROM ,3:TICK
                unset(self::$data_server['alias'][$data[1]]);
                $conn->send([$data[3] ,$data[2],'ok' ]);
                break;
            case 'MONITOR'://0:TOKEN, 1:WATCH_TO, 2:FROM ,3:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $data[1] = (self::$data_server['alias'][$data[1]]);
                }
                if (isset(self::$data_server['coids'][$data[1]])) {
                    self::$data_server['link'][$data[1]][$data[2]]='';
                    $conn->send([$data[3] ,$data[2], 'ok' ]);
                } else {
                    $conn->send([$data[3] ,$data[2], 'err','econnect monitor' ]);
                    //self::$data_server['err'][$conn->id]++;
                }
                break;
            case 'DEMONITOR'://0:TOKEN, 1:WATCH_TO, 2:FROM ,3:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $data[1] = (self::$data_server['alias'][$data[1]]);
                }
                if (isset(self::$data_server['coids'][$data[1]])) {
                    unset(self::$data_server['link'][$data[1]][$data[2]]); 
                    $conn->send([$data[3] ,$data[2], 'ok' ]);
                } else {
                    $conn->send([$data[3] ,$data[2], 'err','econnect demonitor' ]);
                    //self::$data_server['err'][$conn->id]++;
                }
                break;
            case 'LINK'://0:TOKEN, 1:LINK_TO,2:HASH, 3:FROM ,4:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $data[1] = (self::$data_server['alias'][$data[1]]);
                }
                if (isset(self::$data_server['coids'][$data[1]])) { 
                    self::$data_server['link'][$data[1]][$data[3]]=$data[2];
                    self::$data_server['link'][$data[3]][$data[1]]=$data[2];  
                    $conn->send([$data[4] ,$data[3], 'ok' ]);
                } else {
                    $conn->send([$data[4] ,$data[3], 'err','econnect link' ]);
                    //self::$data_server['err'][$conn->id]++;
                }
                break;
            case 'UNLINK'://0:TOKEN, 1:UNLINK, 2:FROM ,3:TICK
                if (isset(self::$data_server['alias'][$data[1]])) {
                    $data[1] = (self::$data_server['alias'][$data[1]]);
                }
                if (isset(self::$data_server['coids'][$data[1]])) {
                    
                    unset(self::$data_server['link'][$data[1]][$data[2]]);
                    unset(self::$data_server['link'][$data[2]][$data[1]]);
 
                    $conn->send([$data[3] ,$data[2], 'ok' ]);
                } else {
                    $conn->send([$data[3] ,$data[2], 'err','econnect unlink' ]);
                    //self::$data_server['err'][$conn->id]++;
                }
                break; 
            default://0:TICK,1:TO,2:CODE,3:MSG
                //print_r($data);
                // print_r( self::$data_server );
                self::$data_server['coids'][$data[1]]->send($data);//callback 
                break;
        }
    }
    public static function onServerClose($conn){
        //clear
        if (isset(self::$data_server['coids'])) {
            foreach (self::$data_server['coids'] as $coid => $option) { 
                if (isset($option) && $option==$conn) { 
                    \call_user_func([self::class,'onServerMessage'], $conn, ['CLOSE',$coid]); 
                }
            }
        }
        echo count(self::$data_server['coids'])."\n";
    }



    public static function onClientMessage($conn, $data)
    {
        //echo \json_encode($data)."\n";
        switch ($data[0]) {
            case 'PING':
                echo '1';
                break;
            case 'NEW':
                $defer = self::$data_server['NEW'][$data[1]]; 
                unset(self::$data_server['NEW'][$data[1]]);
                $defer->resolve($data[2]);
                break;
            case 'SEND':  //0:TOKEN,1:TO,2:MSG ,3:FROM ,4:TICK
                lib\WeakRef::take($data[1])->send($data[1],$data[2] );
                //self::sendTo($data[1], $data[2], $data[4]);
                break;
            case 'EXIT'://0:TOKEN  1:DIE , 2:GEN_TOKEN  ,3:FROM 
                if (isset(self::$data_server['TRAPEXIT'][$data[1]])
                && self::$data_server['TRAPEXIT'][$data[1]] == true ) { 
                    \xlx\lib\WeakRef::take($data[1])->send($data[1],['EXIT',$data[3]  ]);
                    //self::send($data[2], ['EXIT',$data[1] ,$data[3] ] );
                } else { 
                    \xlx\co\abort($data[2]);
                }
                break; 
            default: //0:TICK,1:TO,2:CODE,3:MSG
                //print_r($data);
                if($actor = lib\WeakRef::take($data[1])){
                    $actor->send($data[1],$data );
                } 
                
                //self::sendTo($data[1], $data, $data[0]);
                break;
        }
    }
    public static function onClientClose($conn){
        //file_put_contents(\microtime(true),'aa');
        //self::$conn_server=null;
        exit('<SERVER DIE>'); //自动启动服务器重新链接？？？？
    } 
 

 
    // /**
    //  * 监视服务 
    //  * @return void
    //  */
    // public static function startSupervisor(){   
    //     yield \xlx\AChannel::make($actor,$s);
    //     $actor->register('danboo.sup');
    //     while(1){
    //         $item = yield $actor->receive();
    //         switch ($item[0]) {
    //             case 'spawn'://0:TOKEN 1:FUNC 2:ARGS 3:RETURN
    //                 $coid = ++self::$gen_coids;
    //                 $actor->send($item[3],$coid);//return
    //                 if($item[1] instanceof \Opis\Closure\SSerializableClosure){
    //                     $item[1] = unserialize($item[1]);
    //                 }
    //                 self::$data_server['coids'][$coid] = new \xlx\DbSupervisor($coid,$item); 
    //                 break;
    //             case 'start'://0:TOKEN 1:CLASS_NAME 2:ARGS
    //                 $coid = ++self::$gen_coids;
    //                 if (isset(self::$data_server['alias'][ $item[1] ])) { 
    //                     return;
    //                 }  
    //                 $item[1] = [$item[1],'init'];
    //                 self::$data_server['alias'][ $item[1] ] = $coid;  
    //                 self::$data_server['coids'][$coid] = new \xlx\DbSupervisor($coid,$item);
    //                 break;
    //             case 'start_child'://0:TOKEN 1:COID 2:ARGS 3:RETURN
    //                 $super = self::$data_server['coids'][$item[1]]; 
    //                 $actor->link(yield \xlx\Worker::spawn($super->childs[$item[2]]));
    //                 break;
    //             default:
    //                 # code...
    //                 break;
    //         }
    //     }    
    // }




}